package com.jscloud.bigdata.flink.windows;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * * 滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是 window_size，一个是 sliding_size。
 * * **需求：每收到2个相同 key 的数据就计算一次，每一次计算的 groupwindow 范围是 5 个元素。**
 */
public class CountWindowSlidingByJava {
        public static void main(String[] args) throws Exception {
                StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
                DataStreamSource<String> socketTextStream = environment.socketTextStream("bigdata01", 9999);
                Tuple2<String, Integer> tuple2 = new Tuple2<String, Integer>();
                socketTextStream.flatMap(new FlatMapFunction<String, String>() {
                                @Override
                                public void flatMap(String value, Collector<String> out) throws Exception {
                                        String[] words = value.split(" ");
                                        for (String word : words) {
                                                out.collect(word);
                                        }
                                }
                        }).map(new MapFunction<String, Tuple2<String, Integer>>() {
                                @Override
                                public Tuple2 map(String value) throws Exception {
                                        tuple2.setFields(value, 1);
                                        return tuple2;
                                }
                        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
                                @Override
                                public String getKey(Tuple2<String, Integer> value) throws Exception {
                                        return value.f0;
                                }
                        }).countWindow(5, 2)
                        .sum(1)
                        .print();
                environment.execute();
        }
}